package chapter5.eg2;

import chapter4.eg11.User;
import chapter5.AbstractConsumer;
import java.time.Duration;
import java.util.Collections;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.Test;

public class DeserializerTest extends AbstractConsumer {
  @Test
  public void testDeserialize() {
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "chapter5.eg2.UserDeserializer");
    KafkaConsumer<String, User> consumer = new KafkaConsumer<>(properties);
    consumer.subscribe(Collections.singletonList("test-topic"));
    while (true) {
      ConsumerRecords<String, User> records = consumer.poll(Duration.ofMillis(1000));
      records.forEach(this::printRecord);
    }
  }

}
